package com.hyxt.provider;

import ch.qos.logback.core.net.SyslogOutputStream;
import com.hyxt.DO.AlarmInfo;
import com.hyxt.DO.WarningInfo;
import com.hyxt.utils.DataValue;
import com.hyxt.utils.StringUtil;
import com.hyxt.utils.ValueLatFilters;
import com.hyxt.utils.ValueLonFilters;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.QualifierFilter;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.util.Bytes;
import com.alibaba.dubbo.config.annotation.Service;
import com.hyxt.DO.GPSInfo;
import com.hyxt.api.HbaseCRUD;
import org.apache.hadoop.conf.Configuration;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;


/**
 *
 * @author wxf
 * @version v1.0
 * @Description Hbase操作接口实现类
 * @Date: Create in 12:40 2017/11/23
 * @Modifide By:
 **/
@Service(interfaceClass = HbaseCRUD.class,version = "1.0.0" )
public class HbaseCRUDServiceImpl implements HbaseCRUD {
    // hbase 全局文件
    //private static Configuration conf;
    // hbase 连接类
   //private static HConnection connection;
    // 配置文件
    private static Properties props;

   // static HTableInterface MileageReadTable;
    //日期格式化
    //static DateFormat format = new SimpleDateFormat("yyyyMMddHHmmss");
  //  static HTableInterface DataTable;

  /* static {
        conf = HBaseConfiguration.create();
        try {
            connection = HConnectionManager.createConnection(conf);
            MileageReadTable = connection.getTable("D_MILEAGE");
        } catch (IOException e) {

        }
    }*/
    public HConnection getConnections(){
      //  Configuration con = null;
       // con.addResource("development/hbase-site.xml");

        HConnection connection=null;
        Configuration conf = HBaseConfiguration.create();
        try {
            connection =  HConnectionManager.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return connection;
    }
    /**
     * 增加一条数据
     * */
    static int counter = 0;
    /**
     * 增加一个数据
     * @param tableName
     * @param gpsinfo
     * @return
     * @throws IOException
     */
    @Override
    public int putData(String tableName, GPSInfo gpsinfo) throws IOException {
        DateFormat format = new SimpleDateFormat("yyyyMMddHHmmss");
        // 获取table实例 wxf
        HConnection connection = getConnections();
        HTableInterface DataTable = connection.getTable(tableName);
        // 实例化hbase入库类
        String rowKey = gpsinfo.getVehicleID() + format.format(gpsinfo.getgTime());

        Put putDataValue = new Put(Bytes.toBytes(rowKey));
        // 得到值
        String value = gpsinfo.toHBaseString();
        // 得到列族
        String family = "i";
        // 得到列限定符
        String qualifier = "c";
        // put对象获得所需行健，列族，列限定符，时间戳
        putDataValue.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), System.currentTimeMillis(), Bytes.toBytes(value));
        DataTable.put(putDataValue);
        counter++;
        if (counter >= 12000000) {
            if (counter % 1 == 0) {
                DataTable.flushCommits();
                //log.info("数量:" + counter + "当前车辆:" + gpsinfo.getVehicleID() + "time" + new Date());
            }
        } else {
            if (counter % 10000 == 0) {
                DataTable.flushCommits();
                //log.info("数量:" + counter + "当前车辆:" + gpsinfo.getVehicleID() + "time" + new Date());
            }
        }
        return 0;
    }

    /**
     *
     * @param tableName
     * @param gpsinfo
     * @return
     * @throws Exception
     */
    @Override
    public int putGPSData(String tableName, GPSInfo gpsinfo) throws Exception {
        DateFormat format = new SimpleDateFormat("yyyyMMddHHmmss");
        //获得htable 对象 wxf
        HConnection connections = getConnections();
        // 获取table实例
        //HTableInterface DataTable = HbaseCRUDServiceImpl.connection.getTable(tableName);
       // wxf
        HTableInterface DataTable = connections.getTable(tableName);
        // 实例化hbase入库类
        String rowKey = gpsinfo.getVehicleID() + format.format(gpsinfo.getgTime());

        Put putDataValue = new Put(Bytes.toBytes(rowKey));
        // 得到值
        String value = gpsinfo.toHBaseString();
        // 得到列族
        String family = "i";
        // 得到列限定符
        String qualifier = "c";
        // put对象获得所需行健，列族，列限定符，时间戳
        putDataValue.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), System.currentTimeMillis(), Bytes.toBytes(value));
        DataTable.put(putDataValue);
        counter++;
        try {
            if (counter % 10000 == 0) {
                DataTable.flushCommits();
                //log.info("数量:" + counter + "当前车辆:" + gpsinfo.getVehicleID());
            }
            // table入库

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            DataTable.flushCommits();
            //log.info("最后一次，数量:" + counter + "当前车辆:" + gpsinfo.getVehicleID());
        }
        //log.debug("最后总数:" + counter);
        return 0;
    }

    /**
     *
     * @param tableName
     * @param gpsinfo
     * @return
     * @throws Exception
     */
    @Override
    public int putTestData(String tableName, GPSInfo gpsinfo) throws Exception {
        // 获取table实例
        // HTableInterface DataTable = connection.getTable(tableName);
        DateFormat format = new SimpleDateFormat("yyyyMMddHHmmss");
        //wxf
        HConnection connection = getConnections();
        HTableInterface DataTable = connection.getTable(tableName);
        // 实例化hbase入库类
        String rowKey = gpsinfo.getVehicleID() + format.format(gpsinfo.getgTime());

        Put putDataValue = new Put(Bytes.toBytes(rowKey));
        // 得到值
        String value = gpsinfo.getAddress();
        // 得到列族
        String family = "i";
        // 得到列限定符
        String qualifier = "c";
        // put对象获得所需行健，列族，列限定符，时间戳
        putDataValue.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), System.currentTimeMillis(), Bytes.toBytes(value));
        DataTable.put(putDataValue);
        counter++;

        try {
            if (counter % 3 == 0) {
                DataTable.flushCommits();
                //log.info("数量:" + counter + "当前车辆:" + gpsinfo.getVehicleID());
            }
            // table入库

        } catch (Exception e) {
            e.printStackTrace();
        }
        // log.debug("最后总数:" + counter);
        return 0;
    }

    /**
     * 添加多条数据
     * @param tableName
     * @param rows
     * @return
     */
    @Override
    public int putListData(String tableName, List<DataValue> rows) {

        // 声明Table
        HTableInterface DataTable;
        try {
            // 得到table实例
            //wxf
            HConnection connection = getConnections();
            DataTable = connection.getTable(tableName);
            // 实例化put类型list
            List<Put> putList = new ArrayList<Put>();
            // DataValue 遍历 并设置行健 Put对象 值 列族 列限定符
            for (DataValue dataValue : rows) {
                Put putDataValue = new Put(Bytes.toBytes(dataValue.getRowKey()));
                // 获得值
                String value = (String) dataValue.getDataValue();
                // 获取列族
                String family = dataValue.getFamilyKey();
                // 获取列限定符
                String qualifier = dataValue.getQualifierKey();

                putDataValue.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), dataValue.getTimestaValue(), Bytes.toBytes(value));
                putList.add(putDataValue);
                // log.debug("key" + dataValue.getRowKey());
                // log.debug("正在存入HBase********************************************************");
            }
            DataTable.put(putList);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return 0;
    }

    /**
     * 删除一条数据
     * @param tableName
     * @param rowKey
     * @throws IOException
     */
    @Override
    public void deleteRow(String tableName, String rowKey) throws IOException {
        // table实例
        //wxf
        HConnection connection = getConnections();
        HTableInterface DataTable = connection.getTable(tableName);

        try {
            List<Delete> dataList = new ArrayList<Delete>();
            Delete delData = new Delete(rowKey.getBytes());
            dataList.add(delData);
            DataTable.delete(dataList);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            DataTable.close();
        }
    }

    /**
     * 查询一行数据
     * @param tableName
     * @param rowKey
     * @return
     * @throws IOException
     */
    @Override
    public List<DataValue> selectRow(String tableName, String rowKey) throws IOException {

        List<DataValue> dataList = new ArrayList<DataValue>();
        //wxf
        HConnection connection = getConnections();
        HTableInterface dataTable = connection.getTable(tableName);

        try {
            Get dataGet = new Get(rowKey.getBytes());
            Result r = dataTable.get(dataGet);
            // log.debug("rowKey:" + new String(r.getRow()));
            for (Cell cell : r.rawCells()) {
                // log.debug("--列:" + new String(cell.getFamilyArray())
                // + "--列限定符：" + new String(cell.getQualifierArray()) + "--值:"
                // + new String(cell.getValueArray()));

                DataValue dataValue = new DataValue();
                dataValue.setRowKey(rowKey);
                dataValue.setFamilyKey(String.valueOf(cell.getFamilyArray()));
                dataValue.setQualifierKey(String.valueOf(cell.getQualifierArray()));
                dataValue.setTimestaValue(cell.getTimestamp());
                dataValue.setDataValue(new String(cell.getValueArray()));
                dataList.add(dataValue);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                dataTable.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return dataList;
    }
    @Override
    public String getMileage(String rowKey) throws IOException {

        // HTableInterface dataTable = connection.getTable(tableName);
        String result = null;
        HTableInterface MileageReadTable=null;
        try {
            //wxf
            HConnection connection = getConnections();
            MileageReadTable = connection.getTable("D_MILEAGE");
            //wxf  end
            Get dataGet = new Get(rowKey.getBytes());
            Result r = MileageReadTable.get(dataGet);
            // log.debug("rowKey:" + new String(r.getRow()));
            for (Cell cell : r.rawCells()) {
                // log.debug("--列:" + new String(cell.getFamilyArray())
                // + "--列限定符：" + new String(cell.getQualifierArray()) + "--值:"
                // + new String(cell.getValueArray()));
                result = new String(CellUtil.cloneValue(cell));
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {

        }
        return result;
    }

    /**
     * 查询历史轨迹
     * @param vehicleID 车辆id
     * @param startTime 起始时间
     * @param endTime 结束时间
     * @return
     * @throws IOException
     * @throws java.text.ParseException
     */
    @Override
    public List<GPSInfo> scanHistroyDate(String vehicleID, Date startTime, Date endTime) throws IOException, java.text.ParseException {

        // 创建返回的list
        List<GPSInfo> gpsList = new ArrayList<GPSInfo>();
        // 创建日期格式化
        DateFormat format = new SimpleDateFormat("yyyyMMddHHmmss");
        //wxf
        HConnection connection = getConnections();
       // System.out.println(connection.toString()+"--------------");
        // 根据名称获取Table
        HTableInterface GPSTable = connection.getTable("D_GPSINFO");
       // System.out.println(GPSTable.toString()+"++++++++++++++++++++");
        // 设置起始行健 vid+指定日期格式的日期
        String startRow = vehicleID + format.format(startTime);
        // 设置结束行健 vid+指定日期格式的日期
        String stopRow = vehicleID + format.format(endTime);
        // 初始化HBase 查询对象
        Scan s = new Scan();
        s.setCaching(20160);
        // 分别添加起始/结束行健
        s.setStartRow(startRow.getBytes());
        s.setStopRow(stopRow.getBytes());
        try {
            // 获取查询的结果集
            ResultScanner results = GPSTable.getScanner(s);
            // 遍历结果集
            for (Result r : results) {
                // 遍历结果单元
                for (Cell cell : r.rawCells()) {
                    // 获取查询结果的value
                    String infoStr = new String(CellUtil.cloneValue(cell));
                    // 通过"|" 切分值
                    String[] infoSpl = infoStr.split("\\|");
                    // 分别将对应字段添加
                    GPSInfo gpsInfo = new GPSInfo();
                    // vid
                    gpsInfo.setVehicleID(new String(infoSpl[0]));
                    // 采集时间
                    gpsInfo.setgTime(format.parse((infoSpl[1])));
                    // 经度
                    gpsInfo.setElon(Double.parseDouble(infoSpl[2]));
                    // 纬度
                    gpsInfo.setElat(Double.parseDouble(infoSpl[3]));
                    // 高度
                    gpsInfo.setAlt(Double.parseDouble(infoSpl[4]));
                    // 速度
                    gpsInfo.setSpeed(Float.parseFloat(infoSpl[5]));
                    // 里程
                    gpsInfo.setMileage(Double.parseDouble(infoSpl[6]));

                    gpsInfo.setElevation(Integer.parseInt(infoSpl[7]));
                    // 方向
                    gpsInfo.setDirection(Integer.parseInt(infoSpl[8]));
                    //
                    gpsInfo.setUploadType(Integer.parseInt(infoSpl[9]));
                    // 报警标识
                    gpsInfo.setAlarmTag(Long.valueOf(infoSpl[10]));
                    gpsInfo.setVelstaType(infoSpl[11]);
                    // 加密前经度
                    gpsInfo.setLon(Double.parseDouble(infoSpl[12]));
                    // 加密后纬度
                    gpsInfo.setLat(Double.parseDouble(infoSpl[13]));
                    // 本地信息
                    gpsInfo.setLocalInfo(infoSpl[14]);
                    // 距离
                    gpsInfo.setDistance(Double.parseDouble(infoSpl[15]));
                    gpsInfo.setFuel(Double.parseDouble(infoSpl[16]));
                    // 行驶记录仪记录的速度
                    gpsInfo.setDriveSpeed(Float.parseFloat(infoSpl[17]));
                    // 超速值
                    gpsInfo.setOverspeedValue(Integer.parseInt(infoSpl[18]));
                    // 区域预警值
                    gpsInfo.setAreawarnValue(Integer.parseInt(infoSpl[19]));
                    // 驾驶时间预警值
                    gpsInfo.setDrivetimewarnValue(Integer.parseInt(infoSpl[20]));
                    // 组织编号
                    gpsInfo.setgCode(infoSpl[21]);
                    // 终端类型
                    gpsInfo.setTerminalType(infoSpl[22]);
                    // 地址
                    gpsInfo.setAddress(infoSpl[23]);
                    // 如果是参与计算的点 才拿回集合
                    if (gpsInfo.getAreawarnValue() == 0 || gpsInfo.getAreawarnValue() == 2) {
                        gpsList.add(gpsInfo);
                    }
                }
            }
            // 关闭结果集
            results.close();

        } catch (IOException e) {
            e.printStackTrace();
        }
        return gpsList;
    }
    @Override
    public GPSInfo getLastPoint(String vehicleID, Date startTime, Date endTime) throws IOException, java.text.ParseException {

        // 创建日期格式化
        DateFormat format = new SimpleDateFormat("yyyyMMddHHmmss");
        //wxf
        HConnection connection = getConnections();
        // 根据名称获取Table
        HTableInterface GPSTable = connection.getTable("D_GPSINFO");

        // 设置起始行健 vid+指定日期格式的日期
        String startRow = vehicleID + format.format(startTime);
        // 设置结束行健 vid+指定日期格式的日期
        String stopRow = vehicleID + format.format(endTime);
        // 初始化HBase 查询对象
        Scan s = new Scan();
        s.setCaching(1);
        // 分别添加起始/结束行健
        s.setStartRow(startRow.getBytes());
        s.setStopRow(stopRow.getBytes());
        s.setReversed(true);

        try {
            // 获取查询的结果集
            ResultScanner results = GPSTable.getScanner(s);
            // 遍历结果集
            for (Result r : results) {
                // 遍历结果单元
                for (Cell cell : r.rawCells()) {
                    // 获取查询结果的value
                    String infoStr = new String(CellUtil.cloneValue(cell));
                    // 通过"|" 切分值
                    String[] infoSpl = infoStr.split("\\|");
                    // 分别将对应字段添加
                    GPSInfo gpsInfo = new GPSInfo();
                    // vid
                    gpsInfo.setVehicleID(new String(infoSpl[0]));
                    // 采集时间
                    gpsInfo.setgTime(format.parse((infoSpl[1])));
                    // 经度
                    gpsInfo.setElon(Double.parseDouble(infoSpl[2]));
                    // 纬度
                    gpsInfo.setElat(Double.parseDouble(infoSpl[3]));
                    // 高度
                    gpsInfo.setAlt(Double.parseDouble(infoSpl[4]));
                    // 速度
                    gpsInfo.setSpeed(Float.parseFloat(infoSpl[5]));
                    // 里程
                    gpsInfo.setMileage(Double.parseDouble(infoSpl[6]));
                    gpsInfo.setElevation(Integer.parseInt(infoSpl[7]));
                    // 方向
                    gpsInfo.setDirection(Integer.parseInt(infoSpl[8]));
                    //
                    gpsInfo.setUploadType(Integer.parseInt(infoSpl[9]));
                    // 报警标识
                    gpsInfo.setAlarmTag(Long.valueOf(infoSpl[10]));
                    gpsInfo.setVelstaType(infoSpl[11]);
                    // 加密前经度
                    gpsInfo.setLon(Double.parseDouble(infoSpl[12]));
                    // 加密后纬度
                    gpsInfo.setLat(Double.parseDouble(infoSpl[13]));
                    // 本地信息
                    gpsInfo.setLocalInfo(infoSpl[14]);
                    // 距离
                    gpsInfo.setDistance(Double.parseDouble(infoSpl[15]));
                    gpsInfo.setFuel(Double.parseDouble(infoSpl[16]));
                    // 行驶记录仪记录的速度
                    gpsInfo.setDriveSpeed(Float.parseFloat(infoSpl[17]));
                    // 超速值
                    gpsInfo.setOverspeedValue(Integer.parseInt(infoSpl[18]));
                    // 区域预警值
                    gpsInfo.setAreawarnValue(Integer.parseInt(infoSpl[19]));
                    // 驾驶时间预警值
                    gpsInfo.setDrivetimewarnValue(Integer.parseInt(infoSpl[20]));
                    // 组织编号
                    gpsInfo.setgCode(infoSpl[21]);
                    // 终端类型
                    gpsInfo.setTerminalType(infoSpl[22]);

                    if (gpsInfo.getAreawarnValue() == 0) {
                        //log.debug(gpsInfo.getgTime());
                        results.close();
                        return gpsInfo;
                    }
                }
            }
            // 关闭结果集
            results.close();

        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    @Override
    public GPSInfo getLastPointByID(String vid, Date startTime, Date endTime) {
        Map<String, GPSInfo> map = new HashMap<String, GPSInfo>();
        String vihecleID = vid;
        try {
            List<GPSInfo> list = scanHistroyDate(vid, startTime, endTime);
            if (list.size() == 0) {
                return null;
            }
            for (GPSInfo gpsInfo : list) {
                if (gpsInfo.getAreawarnValue() == 1 || gpsInfo.getAreawarnValue() == 2 || gpsInfo.getAreawarnValue() == 3) {
                    continue;
                }
                map.put(vid, gpsInfo);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (ParseException e) {
            e.printStackTrace();
        }
        return map.get(vihecleID);
    }
    @Override
    public double getDayMileageByVid(String vid, Date startTime, Date endTime) {
        Map<String, Double> map = new HashMap<String, Double>();
        try {
            List<GPSInfo> list = scanHistroyDate(vid, startTime, endTime);
            if (list.size() == 0) {
                return -1;
            }
            for (GPSInfo gpsInfo : list) {
                if (gpsInfo.getAreawarnValue() == 0 && (!map.containsKey("1"))) {
                    map.put("1", gpsInfo.getMileage());
                }
                if (gpsInfo.getAreawarnValue() == 0) {
                    map.put(vid, gpsInfo.getMileage());
                }
            }
            if (map.get("1") == null) {
                return -1;
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (ParseException e) {
            e.printStackTrace();
        }
        double first_mileage = map.get("1");
        double last_mileage = map.get(vid);
        return (last_mileage - first_mileage);
    }

    /**
     * 历史报警查询
     * @param vehicleID
     * @param startTime
     * @param endTime
     * @return
     * @throws IOException
     * @throws java.text.ParseException
     */
    @Override
    public List<AlarmInfo> scanAlarmHistroyDate(String vehicleID, Date startTime, Date endTime) throws IOException, java.text.ParseException {
        List<AlarmInfo> alarmList = new ArrayList<AlarmInfo>();
        // 时间特定格式 yyyyMMddHHmmss
        DateFormat format = new SimpleDateFormat("yyyyMMddHHmmss");
        //wxf
        HConnection connection = getConnections();
        // 获取table实例
        HTableInterface AlarmTable = connection.getTable("D_ALARMDETAIL");

        // 设置行健 起始行健 vid+特定时间格式+"000" 模糊查询以000起始
        String startRow = vehicleID + format.format(startTime) + "000";
        // 设置行健 结束行健 vid+特定时间格式+"999" 模糊查询以999结尾
        String stopRow = vehicleID + format.format(endTime) + "999";
        // 获得查询实例
        Scan s = new Scan();
        // 设置查询缓存大小
        s.setCaching(20160);
        // 添加起始结束键
        s.setStartRow(startRow.getBytes());
        s.setStopRow(stopRow.getBytes());

        try {
            // 查询并范围结果集
            ResultScanner results = AlarmTable.getScanner(s);
            // 迭代结果
            for (Result r : results) {
                // 迭代具体单元
                for (Cell cell : r.rawCells()) {
                    // 得到具体值value
                    String infoStr = new String(CellUtil.cloneValue(cell));
                    // 以分割符"|"分割目标值,得到所需要的值的String[]
                    String[] infoSpl = infoStr.split("\\|");
                    // 实例报警信息实体类
                    AlarmInfo alarmInfo = new AlarmInfo();

                    alarmInfo.setVehicleID(new String(infoSpl[0])); // 车ID
                    alarmInfo.setgTime(format.parse((infoSpl[1]))); // 采集数据的gps时间
                    alarmInfo.setElon(Double.parseDouble(infoSpl[2])); // 加密后经度
                    alarmInfo.setElat(Double.parseDouble(infoSpl[3])); // 加密后纬度
                    alarmInfo.setAlt(Double.parseDouble(infoSpl[4])); // 高度
                    alarmInfo.setSpeed(Float.parseFloat(infoSpl[5])); // 速度
                    alarmInfo.setMileage(Double.parseDouble(infoSpl[6])); // 里程
                    alarmInfo.setElevation(Integer.parseInt(infoSpl[7]));
                    alarmInfo.setDirection(Integer.parseInt(infoSpl[8])); // 距离
                    alarmInfo.setUploadType(Integer.parseInt(infoSpl[9])); // 上传类型
                    alarmInfo.setAlarmType(infoSpl[10]); // 报警类型
                    alarmInfo.setVelstaType(infoSpl[11]);
                    alarmInfo.setLon(Double.parseDouble(infoSpl[12])); // 加密前经度
                    alarmInfo.setLat(Double.parseDouble(infoSpl[13])); // 加密前经度
                    alarmInfo.setLocalInfo(infoSpl[14]); // 本地信息
                    alarmInfo.setDistance(Double.parseDouble(infoSpl[15])); // 距离
                    alarmInfo.setFuel(Double.parseDouble(infoSpl[16])); // 油量
                    alarmInfo.setDriveSpeed(Float.parseFloat(infoSpl[17])); // 行驶记录仪速度
                    alarmInfo.setOverspeedValue(Integer.parseInt(infoSpl[18])); // 超速预警值
                    alarmInfo.setAreawarnValue(Integer.parseInt(infoSpl[19])); // 区域预警值
                    alarmInfo.setDrivetimewarnValue(Integer.parseInt(infoSpl[20])); // 行驶时间预警值
                    alarmInfo.setgCode(infoSpl[21]); // 组织编号
                    alarmInfo.setTerminalType(infoSpl[22]); // 终端类型

                    alarmList.add(alarmInfo);

                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            AlarmTable.close();// 查询后关闭表实例
        }
        return alarmList;
    }

    /**
     * 查询历史预警信息
     * @param vehicleID 车辆id
     * @param startTime 开始时间
     * @param endTime 结束时间
     * @return
     * @throws IOException
     * @throws java.text.ParseException
     */
    @Override
    public List<WarningInfo> scanWarnHistroyDate(String vehicleID, Date startTime, Date endTime) throws IOException, java.text.ParseException {
        // 创建预警信息集合
        List<WarningInfo> warningList = new ArrayList<WarningInfo>();
        // 设置特定时间格式
        DateFormat format = new SimpleDateFormat("yyyyMMddHHmmss");
        //wxf
        HConnection connection = getConnections();
        // 获取预警表实例
        HTableInterface WarnTable = connection.getTable("D_WARNINGDETAIL");

        // 创建起始行健 车ID + 特定时间格式 + "000" (3位预警值,模糊查询)
        String startRow = vehicleID + format.format(startTime) + "000";
        // 创建结束行健 车ID + 特定时间格式 + "999" (3位预警值,模糊查询)
        String stopRow = vehicleID + format.format(endTime) + "999";
        // 获取扫描查询实例
        Scan s = new Scan();
        // 设置客户端查询缓存大小
        s.setCaching(20160);
        // 加入起始/结束行健,卡两头rowkey
        s.setStartRow(startRow.getBytes());
        s.setStopRow(stopRow.getBytes());

        try {
            // 获得结果集
            ResultScanner results = WarnTable.getScanner(s);
            // 迭代结果
            for (Result r : results) {
                // 迭代值的单元
                for (Cell cell : r.rawCells()) {
                    // 迭代单元值,获得所需要的值value
                    String infoStr = new String(CellUtil.cloneValue(cell));
                    // 将目标值通过分隔符"|",切分为String[]
                    String[] infoSpl = infoStr.split("\\|");
                    // 实例预警信息实例
                    WarningInfo warningInfo = new WarningInfo();

                    warningInfo.setVehicleID(new String(infoSpl[0])); // 车ID
                    warningInfo.setgTime(format.parse((infoSpl[1]))); // 获取GPS时间
                    warningInfo.setElon(Double.parseDouble(infoSpl[2])); // 加密后经度
                    warningInfo.setElat(Double.parseDouble(infoSpl[3])); // 加密后纬度
                    warningInfo.setAlt(Double.parseDouble(infoSpl[4])); // 高度
                    warningInfo.setSpeed(Float.parseFloat(infoSpl[5])); // 速度
                    warningInfo.setMileage(Double.parseDouble(infoSpl[6])); // 里程
                    warningInfo.setElevation(Integer.parseInt(infoSpl[7])); //
                    warningInfo.setDirection(Integer.parseInt(infoSpl[8])); // 方向
                    warningInfo.setUploadType(Integer.parseInt(infoSpl[9])); // 上传类型
                    warningInfo.setWarnType(infoSpl[10]); // 预警类型
                    warningInfo.setVelstaType(infoSpl[11]); //
                    warningInfo.setLon(Double.parseDouble(infoSpl[12])); // 加密前经度
                    warningInfo.setLat(Double.parseDouble(infoSpl[13])); // 加密后纬度
                    warningInfo.setLocalInfo(infoSpl[14]); // 本地信息
                    warningInfo.setDistance(Double.parseDouble(infoSpl[15])); // 距离
                    warningInfo.setFuel(Double.parseDouble(infoSpl[16])); // 油量
                    warningInfo.setDriveSpeed(Float.parseFloat(infoSpl[17])); // 行驶记录仪速度
                    warningInfo.setOverspeedValue(Integer.parseInt(infoSpl[18])); //
                    warningInfo.setAreawarnValue(Integer.parseInt(infoSpl[19]));
                    warningInfo.setDrivetimewarnValue(Integer.parseInt(infoSpl[20]));
                    warningInfo.setgCode(infoSpl[21]); // 组织编号
                    warningInfo.setTerminalType(infoSpl[22]); // 终端类型
                    warningList.add(warningInfo);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            WarnTable.close(); // 查询后关闭获取的表实例
        }
        return warningList;
    }

    /**
     * 定时定位查询
     * @param startTime 开始时间
     * @param endTime 结束时间
     * @param minLon  最小经度
     * @param minLat 最小纬度
     * @param maxLon 最大经度
     * @param maxLat 最大纬度
     * @return
     * @throws java.text.ParseException
     * @throws IOException
     */
    @Override
    public List<GPSInfo> scanAreaQuery(Date startTime, Date endTime, double minLon, double minLat, double maxLon, double maxLat) throws java.text.ParseException, IOException {
        // 实例化Map,为了最后保留结果集中每一辆车的最后一条信息
        HashMap<String, GPSInfo> map = new HashMap<String, GPSInfo>();
        // 实例化目标返回值list
        List<GPSInfo> resultList = new ArrayList<GPSInfo>();
        int intMinlon = (int) minLon; // 取得最小经度的整数
        int intMaxlon = (int) maxLon; // 取得最大经度的整数
        int intMinlat = (int) (minLat * 100); // 取得最小纬度*100
        int intMaxlat = (int) (maxLat * 100); // 取得最大纬度*100

        List<GPSInfo> gpsList = new ArrayList<GPSInfo>();
        // 设置目标时间格式1,此格式应用于行健中
        DateFormat format = new SimpleDateFormat("yyyyMMddHH");
        // 设置目标时间格式2,此格式应用于值,即value中
        DateFormat format2 = new SimpleDateFormat("yyyyMMddHHmmss");
        //wxf
        HConnection connection = getConnections();
        // 获取区域查车表的实例
        HTableInterface AreaQueryTable = connection.getTable("D_AREAQUERY");

        // 创建起始行健 给定时间格式(yyyyMMddHH) + 经度取整数部分3位(不足3位左补0) +
        // （纬度*100）取整数部分4位,位不足左补0 + 00000000
        String startRow = format.format(startTime) + StringUtil.padding(intMinlon + "", "0", 3) + StringUtil.padding(intMinlat + "", "0", 4) + "00000000";
        // 创建结束行健 给定时间格式(yyyyMMddHH) + 经度取整数部分3位(不足3位左补0) +
        // （纬度*100）取整数部分4位,位不足左补0 + 99999999
        String stopRow = format.format(endTime) + StringUtil.padding(intMaxlon + "", "0", 3) + StringUtil.padding(intMaxlat + "", "0", 4) + "99999999";

        // 定义过滤器,为了过滤不需要的数据
        FilterList filtList = new FilterList();
        // 取列限定符范围 大于等于最小列限定符 小于等于最大列限定符
        Filter filt_Greater = new QualifierFilter(CompareOp.GREATER_OR_EQUAL, new BinaryComparator(Bytes.toBytes("00")));
        Filter filt_Less = new QualifierFilter(CompareOp.LESS_OR_EQUAL, new BinaryComparator(Bytes.toBytes("99")));

        // 采用自定义过滤比较器 目标值的经度 >= 最小经度 <=最大经度 目标值纬度 >= 最小纬度 <=最大纬度
        Filter lon_min = new ValueLonFilters(CompareOp.GREATER_OR_EQUAL, new BinaryComparator(Bytes.toBytes(minLon)));
        Filter lon_max = new ValueLonFilters(CompareOp.LESS_OR_EQUAL, new BinaryComparator(Bytes.toBytes(intMaxlon)));
        Filter lat_min = new ValueLatFilters(CompareOp.GREATER_OR_EQUAL, new BinaryComparator(Bytes.toBytes(minLat)));
        Filter lat_max = new ValueLatFilters(CompareOp.LESS_OR_EQUAL, new BinaryComparator(Bytes.toBytes(maxLat)));

        // 添加过滤器们
        filtList.addFilter(filt_Greater);
        filtList.addFilter(filt_Less);
        filtList.addFilter(lon_min);
        filtList.addFilter(lon_max);
        filtList.addFilter(lat_min);
        filtList.addFilter(lat_max);

        Scan s = new Scan();
        s.setCaching(30000);
        // 设置起始结束行健
        s.setStartRow(startRow.getBytes());
        s.setStopRow(stopRow.getBytes());
        // 这是过滤器
        s.setFilter(filtList);

        try {
            // 查询返回结果集
            ResultScanner results = AreaQueryTable.getScanner(s);
            // 迭代结果
            for (Result r : results) {
                // 迭代单元
                for (Cell cell : r.rawCells()) {
                    // 获取单元中的值 value
                    String infoStr = new String(CellUtil.cloneValue(cell));
                    // 将目标值通过分隔符"|"切分为GPS信息的String[]
                    String[] infoSpl = infoStr.split("\\|");
                    GPSInfo gpsInfo = new GPSInfo();
                    gpsInfo.setVehicleID(new String(infoSpl[0])); // 车ID
                    gpsInfo.setgTime(format2.parse((infoSpl[1])));// 采集的gps时间
                    gpsInfo.setElon(Double.parseDouble(infoSpl[2])); // 经度
                    gpsInfo.setElat(Double.parseDouble(infoSpl[3])); // 纬度
                    gpsInfo.setAlt(Double.parseDouble(infoSpl[4]));// 高度
                    gpsInfo.setSpeed(Float.parseFloat(infoSpl[5]));// 速度
                    gpsInfo.setMileage(Double.parseDouble(infoSpl[6]));// 里程
                    gpsInfo.setElevation(Integer.parseInt(infoSpl[7]));//
                    gpsInfo.setDirection(Integer.parseInt(infoSpl[8]));// 方向
                    gpsInfo.setUploadType(Integer.parseInt(infoSpl[9])); // 上传类型
                    gpsInfo.setAlarmTag(Long.valueOf(infoSpl[10])); // 报警类型
                    gpsInfo.setVelstaType(infoSpl[11]); //
                    gpsInfo.setLon(Double.parseDouble(infoSpl[12])); // 加密前经度
                    gpsInfo.setLat(Double.parseDouble(infoSpl[13])); // 加密前纬度
                    gpsInfo.setLocalInfo(infoSpl[14]); // 本地信息
                    gpsInfo.setDistance(Double.parseDouble(infoSpl[15])); // 距离
                    gpsInfo.setFuel(Double.parseDouble(infoSpl[16])); //
                    gpsInfo.setDriveSpeed(Float.parseFloat(infoSpl[17])); // 行驶记录仪显示速度
                    gpsInfo.setOverspeedValue(Integer.parseInt(infoSpl[18]));//
                    gpsInfo.setAreawarnValue(Integer.parseInt(infoSpl[19])); //
                    gpsInfo.setDrivetimewarnValue(Integer.parseInt(infoSpl[20])); //
                    gpsInfo.setgCode(infoSpl[21]); // 组织编号
                    gpsInfo.setTerminalType(infoSpl[22]); // 终端类型
                    gpsList.add(gpsInfo);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        try {
            // 关闭表实例
            AreaQueryTable.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 遍历GPSInfo信息,进行一次本地过滤 在最大/最小经纬度之间的信息
        for (GPSInfo gpsInfo : gpsList) {
            if (gpsInfo.getElon() <= maxLon && gpsInfo.getElat() <= maxLat) {
                map.put(gpsInfo.getVehicleID(), gpsInfo);
            }
        }
        // 因为最后要获取每辆车的最后一次GPSInfo信息,以车作为Map的key,信息为value,得到的list为最后想获取的信息
        for (Map.Entry<String, GPSInfo> entry : map.entrySet()) {
            resultList.add(entry.getValue());
        }
        return resultList;
    }

    /**
     *
     * @param orgcode
     * @param startTime
     * @param endTime
     * @throws IOException
     */
    @Override
    public void deleteOldData(String orgcode, Date startTime, Date endTime) throws IOException {
        List<Delete> delList = new ArrayList<Delete>();

        DateFormat format = new SimpleDateFormat("yyyyMMddHHmmss");
        //wxf
        HConnection connection = getConnections();
        HTableInterface LOGTable = connection.getTable("D_LOG");

        String startRow = orgcode + format.format(startTime);
        String stopRow = orgcode + format.format(endTime);
        Scan s = new Scan();
        s.setCaching(20160);
        s.setStartRow(startRow.getBytes());
        s.setStopRow(stopRow.getBytes());

        try {
            ResultScanner results = LOGTable.getScanner(s);
            for (Result r : results) {
                for (Cell cell : r.rawCells()) {
                    Delete d = new Delete(CellUtil.cloneRow(cell));
                    delList.add(d);
                }
            }
            results.close();
            LOGTable.delete(delList);
            LOGTable.close();

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     *
     * @param filename
     * @param buff
     * @return
     * @throws IOException
     */
    @Override
    public int loadFileToHBase(String filename, byte[] buff) throws IOException {
        HTableInterface DataTable = null;
        try {
            //wxf
            HConnection connection = getConnections();
            DataTable = connection.getTable("D_FILE");
            Put put = new Put(filename.getBytes());
            put.add("i".getBytes(), "c".getBytes(), System.currentTimeMillis(), buff);
            DataTable.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            DataTable.close();
        }
        return 1;
    }

    /**
     *
     * @param filename
     * @return
     * @throws IOException
     */
    @Override
    public byte[] findFileFromHBase(String filename) throws IOException {
        HTableInterface dataTable = null;
        byte[] results = null;
        try {
            //wxf
            HConnection connection = getConnections();
            dataTable = connection.getTable("D_FILE");
            Get dataget = new Get(filename.getBytes());
            Result r = dataTable.get(dataget);
            for (Cell cell : r.rawCells()) {
                results = CellUtil.cloneValue(cell);
            }

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            dataTable.close();
        }
        return results;
    }
    @Override
    public List<GPSInfo> scanHistroyDateWriteFile(String vehicleID, Date startTime, Date endTime) throws IOException, java.text.ParseException {
        List<GPSInfo> gpsList = new ArrayList<GPSInfo>();

        DateFormat format = new SimpleDateFormat("yyyyMMddHHmmss");
        DateFormat format2 = new SimpleDateFormat("yyyyMMdd");
        //wxf
        HConnection connection = getConnections();
        HTableInterface GPSTable = connection.getTable("D_GPSINFO");

        String startRow = vehicleID + format.format(startTime);
        String stopRow = vehicleID + format.format(endTime);
        Scan s = new Scan();
        s.setCaching(40160);
        s.setStartRow(startRow.getBytes());
        s.setStopRow(stopRow.getBytes());

        File file = new File("/Users/jakiro/Documents/hdata/" + vehicleID + ".txt");

        if (!file.exists()) {
            file.createNewFile();
        }
        FileWriter writer = new FileWriter(file, false);
        try {
            ResultScanner results = GPSTable.getScanner(s);
            for (Result r : results) {
                for (Cell cell : r.rawCells()) {
                    String infoStr = new String(CellUtil.cloneValue(cell));
                    // log.debug(infoStr);
                    // String[] infoSpl = infoStr.split("\\|");
                    // writer.write(infoSpl[0]+";"); //vid
                    // writer.write(infoSpl[2]+";");//经度
                    // writer.write(infoSpl[3]+";");//纬度
                    // writer.write(String.valueOf(format.parse((infoSpl[1])).getTime())+";");//时间
                    // writer.write(infoSpl[6]);//里程
                    writer.write(infoStr);
                    writer.write("\n");

                    // GPSInfo gpsInfo = new GPSInfo();

                    // gpsList.add(gpsInfo);
                    writer.flush();
                }
            }

            writer.close();
            results.close();

        } catch (IOException e) {
            e.printStackTrace();
        }
        return gpsList;
    }



    public void init(Map<String, ?> dependences) {

    }
    /**
     * 无参构造方法
     * */
    public HbaseCRUDServiceImpl() {

    }
    /**
     * connection获取方法
     * */
    public HConnection getConnection() {
        //wxf
        HConnection connection = getConnections();
        return connection;
    }
    // 获取配置
    public Properties getProperties() {
        return props;
    }
    // 设置配置
    public void setProperties(Properties properties) {
        //wxf
        Configuration conf = HBaseConfiguration.create();
        props = properties;
        // 创建HBase全文文件
        conf = HBaseConfiguration.create();
        // 读取文件设置zookeeper地址
        conf.set("hbase.zookeeper.quorum", props.getProperty("hbase.zookeeper.quorum"));
        // 读取文件设置zookeeper端口
        conf.set("hbase.zookeeper.property.clientPort", props.getProperty("hbase.zookeeper.property.clientPort"));
        // 读取文件设置写入buffer大小
        conf.set("hbase.client.write.buffer", props.getProperty("hbase.client.write.buffer"));
        // 读取文件设置客户端访问重试次数
        conf.set("hbase.client.retries.number", props.getProperty("hbase.client.retries.number"));
        // 读取文件设置读取数据最大大小
        conf.set("hbase.client.keyvalue.maxsize", props.getProperty("hbase.client.keyvalue.maxsize"));
        // 读取文件设置hbase rpc超时时间
        conf.set("hbase.rpc.timeout", props.getProperty("hbase.rpc.timeout"));
        // 读取文件配置hbase数据所在目录
        conf.set("hbase.rootdir", props.getProperty("hbase.rootdir"));
        // 读取文件设置重试休眠时间
        conf.set("hbase.client.pause", props.getProperty("hbase.client.pause"));
        // 读取文件设置是否打开快照许可
        conf.set("hbase.snapshot.enabled", props.getProperty("hbase.snapshot.enabled"));
        // 读取配置文件设置是否启用kerberos做安全验证
        conf.set("hbase.security.authentication", props.getProperty("hbase.security.authentication"));
        // 读取配置文件设置zookeeper使用目录
        conf.set("zookeeper.znode.parent", props.getProperty("zookeeper.znode.parent"));
        // 读取配置设置设置rootserver
        conf.set("zookeeper.znode.rootserver", props.getProperty("zookeeper.znode.rootserver"));
        // 读取配置文件设置zookeeper session超时时间
        conf.set("zookeeper.session.timeout", props.getProperty("zookeeper.session.timeout"));
        // "hbase.dynamic.jars.dir", "hdfs://master.hadoop.com:8020/hbase/jars"
        // conf.set("hbase.dynamic.jars.dir",
        // props.getProperty("hbase.dynamic.jars.dir"));
        conf.set("hbase.master.port", props.getProperty("hbase.master.port"));

        conf.set("hbase.regionserver.port", props.getProperty("hbase.regionserver.port"));
        try {
            //wxf
            HConnection connection = getConnections();
            // 创建连接实例
            connection = HConnectionManager.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
